package com.cjkj.equipment.handler;

import com.cjkj.common.core.exception.ServiceException;
import com.cjkj.equipment.config.AMQPConfig;
import com.cjkj.equipment.service.AMQPService;
import com.cjkj.equipment.service.TJ03DataService;
import com.cjkj.equipment.utils.AMQPUtils;
import lombok.extern.slf4j.Slf4j;
import org.springframework.stereotype.Service;

import javax.annotation.Resource;
import javax.jms.Message;
import javax.jms.MessageListener;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;

/**
 * 多参数监测仪数据处理类
 */
@Service
@Slf4j
public class TJ03Handler implements AMQPService {

    @Resource
    AMQPConfig amqp;

    @Resource
    private TJ03DataService dataService;

    @Override
    public void initAMQP() {
        try {
            AMQPUtils.init(amqp.getTj03GroupId(), amqp, messageListener);
        }catch (Exception e){
            log.error("初始化连接失败：",e);
        }
    }

    private final static ExecutorService executorService = new ThreadPoolExecutor(
            Runtime.getRuntime().availableProcessors(),
            Runtime.getRuntime().availableProcessors() * 2, 60, TimeUnit.SECONDS,
            new LinkedBlockingQueue<>(50000));

    private final MessageListener messageListener = new MessageListener() {
        @Override
        public void onMessage(final Message message) {
            try {
                //1.收到消息之后一定要ACK。
                // 推荐做法：创建Session选择Session.AUTO_ACKNOWLEDGE，这里会自动ACK。
                // 其他做法：创建Session选择Session.CLIENT_ACKNOWLEDGE，这里一定要调message.acknowledge()来ACK。
                // message.acknowledge();
                //2.建议异步处理收到的消息，确保onMessage函数里没有耗时逻辑。
                // 如果业务处理耗时过程过长阻塞住线程，可能会影响SDK收到消息后的正常回调。
                executorService.submit(() -> dataService.processMessage(message));
            } catch (Exception e) {
                log.error("submit task occurs exception ", e);
                throw new ServiceException(e.getMessage());
            }
        }
    };
}